package iharbor2

import (
	"bytes"
	"context"
	"crypto/md5"
	"encoding/hex"
	"github.com/rclone/rclone/backend/iharbor2/api"
	"github.com/rclone/rclone/fs"
	"github.com/rclone/rclone/lib/readers"
	"github.com/rclone/rclone/lib/rest"
	"golang.org/x/sync/errgroup"
	"io"
	"regexp"
	"sync"
)

// 分片上传 无并发操作
func (f *Fs) fileMultipartUpload(ctx context.Context, o *Object, in io.Reader, src fs.ObjectInfo, chunkSize fs.SizeSuffix) (error error, md5string string) {
	var (
		reset     string
		finished  = false
		position  = int64(0)
		remaining = src.Size()
		hashmd5v1 = md5.New()
	)

	for part := int64(0); !finished; part++ {
		n := int64(chunkSize)
		if remaining > 0 {

			if remaining < n {
				n = remaining // 最后的块大小
			}
			// 将 in 分片
			//seg := readers.NewRepeatableReader(io.LimitReader(in, n))
			//MD5 := f.ReaderMd5(ctx, seg)
			//fmt.Println("第 %d 块， md5值 %s", part, MD5)
			//
			//_, _ = seg.Seek(0, io.SeekStart)
			buf := make([]byte, n)
			buf = buf[:n]
			_, err := readers.ReadFill(in, buf)
			if err == io.EOF {
				if n == 0 && part != 0 { // end if no data and if not first chunk
					break
				}
			} else if err != nil {
				return err, ""
			}
			if part == 0 {
				reset = "true"
			} else {
				reset = "false"
			}
			io.Copy(hashmd5v1, bytes.NewBuffer(buf))

			if f.opt.Version == "v1" {
				err := f.v1UploadFile(ctx, o, buf, position, n, part, src.String(), reset)
				if err != nil {
					finished = true
				}
			}
			if f.opt.Version == "v2" {
				err := f.v2RequestHttp(ctx, o, buf, reset, "", part, position, n)
				if err != nil {
					finished = true // 循环结束
				}
			}

		} else {
			break
		}
		remaining -= n
		position += n
	}

	return nil, hex.EncodeToString(hashmd5v1.Sum(nil))
}

// V1EmptyFileUpdate v1接口上传空文件
func (f *Fs) V1EmptyFileUpdate(ctx context.Context, o *Object, in io.Reader, lengsize int64) error {
	bucket, bucketPath := o.split()
	var opts rest.Opts
	Path := "/api/" + defaultVersion + "/metadata/" + bucket + "/" + bucketPath + "/"
	opts = f.HandlerAccesskeyHeader("POST", Path, 3600, nil, &lengsize, nil, in)
	//
	//opts := rest.Opts{
	//	Method:       "POST",
	//	RootURL:      f.opt.Endpoint + "/api/" + defaultVersion + "/metadata/",
	//	Path:         bucket + "/" + bucketPath + "/",
	//	Body:         in,
	//	ExtraHeaders: map[string]string{
	//		//"Authorization": "Token " + f.info.Token,
	//		//"Content-MD5":   md5Str,
	//	},
	//	ContentLength: &lengsize,
	//}
	var response api.V1EmptyFileUpload

	err := o.fs.pacer.CallNoRetry(func() (bool, error) {
		resp, err := f.srv.CallJSON(ctx, &opts, nil, &response)
		retry, err := f.shouldRetry(ctx, resp, err)
		if err != nil {
			fs.Errorf(o, "Failed to upload empty file: %d: %#v", retry, err)
		}

		return false, err
	})
	if err != nil {
		fs.Errorf(o, "空文件上传失败：%s", bucketPath)
		return err
	}
	return nil
}

// 分片上传 并发操作
func (f *Fs) fileMultipartUpload2(ctx context.Context, o *Object, in io.Reader, src fs.ObjectInfo, chunkSize fs.SizeSuffix) (err error, md5string string) {

	var (
		wg        sync.WaitGroup
		finished  = false
		ch        = make(chan struct{}, f.opt.NumberOfConcurrent) // 控制并发数量
		position  = int64(0)                                      // 偏移量
		remaining = src.Size()
		hashmd5v2 = md5.New()
		_, gCtx   = errgroup.WithContext(ctx)
		mu        sync.Mutex
		errupload error
	)

	defer close(ch)

	for part := int64(0); !finished; part++ {

		n := int64(chunkSize)
		if remaining > 0 {
			if remaining < n {
				n = remaining // 最后的块大小
				finished = true
			}
			// 将 in 分片
			//seg := readers.NewRepeatableReader(io.LimitReader(in, n))
			//_, _ = seg.Seek(0, io.SeekStart)

			buf := make([]byte, n)

			if gCtx.Err() != nil {
				buf = buf[:0]
				break
			}

			_, err = readers.ReadFill(in, buf)
			if err == io.EOF {
				if n == 0 && part != 0 { // end if no data and if not first chunk
					buf = buf[:0]
					break
				}
			} else if err != nil {
				buf = buf[:0]
				return err, ""
			}
			mu.Lock()
			buf = buf[:n]
			io.Copy(hashmd5v2, bytes.NewBuffer(buf))
			mu.Unlock()

			if f.opt.Version == "v1" {
				if part != 0 {
					ch <- struct{}{}
					wg.Add(1)
					go func(part, position int64, buf []byte) {
						defer func() {
							buf = buf[:0]
							wg.Done()
						}()
						err = f.v1UploadFile(ctx, o, buf, position, n, part, src.String(), "false")
						if err != nil {
							finished = true // 结束循环
							errupload = err
						}
						<-ch
					}(part, position, buf)
				} else {
					err = f.v1UploadFile(ctx, o, buf, position, n, part, src.String(), "true")
					if err != nil {
						finished = true // 结束循环
						errupload = err
					}
				}
			}

			if f.opt.Version == "v2" {
				if part != 0 {
					ch <- struct{}{}
					wg.Add(1)
					go func(part, position int64, buf []byte) {
						defer func() {
							buf = buf[:0]
							wg.Done()
						}()
						err = f.v2RequestHttp(ctx, o, buf, "false", "", part, position, n)
						if err != nil {
							finished = true // 结束循环
							errupload = err
						}
						<-ch
					}(part, position, buf)
				} else {
					err = f.v2RequestHttp(ctx, o, buf, "true", "", part, position, n)
					if err != nil {
						finished = true // 结束循环
						errupload = err
					}
				}
			}

		} else {
			break
		}
		remaining -= n
		position += n

	}
	wg.Wait()
	if err != nil {
		matchString, _ := regexp.MatchString("上传失败", err.Error())
		if matchString == true {
			return err, ""
		}

	}
	if errupload != nil {
		return errupload, hex.EncodeToString(hashmd5v2.Sum(nil))
	}

	return nil, hex.EncodeToString(hashmd5v2.Sum(nil))

}

//func (f *Fs) fileMultipartUpload4(ctx context.Context, o *Object, in io.Reader, src fs.ObjectInfo, chunkSize fs.SizeSuffix) (error error, md5string string) {
//
//	var (
//		wg        sync.WaitGroup
//		finished  = false
//		ch        = make(chan struct{}, f.opt.NumberOfConcurrent) // 控制并发数量
//		position  = int64(0)                                      // 偏移量
//		remaining = src.Size()
//		hashmd5v2 = md5.New()
//		//g, gCtx  = errgroup.WithContext(ctx)
//	)
//
//	defer close(ch)
//
//	for part := int64(0); !finished; part++ {
//
//		n := int64(chunkSize)
//		if remaining > 0 {
//			if remaining < n {
//				n = remaining // 最后的块大小
//				finished = true
//			}
//			// 将 in 分片
//			//seg := readers.NewRepeatableReader(io.LimitReader(in, n))
//			//_, _ = seg.Seek(0, io.SeekStart)
//
//			buf := make([]byte, n)
//			buf = buf[:n]
//			_, err := readers.ReadFill(in, buf)
//			if err == io.EOF {
//				if n == 0 && part != 0 { // end if no data and if not first chunk
//					defer atexit.OnError(&err, func() {})()
//					break
//				}
//			} else if err != nil {
//				return err, ""
//			}
//			io.Copy(hashmd5v2, bytes.NewBuffer(buf))
//
//			if f.opt.Version == "v1" {
//				if part != 0 {
//					ch <- struct{}{}
//					wg.Add(1)
//					go func(part, position int64, buf []byte) {
//						defer wg.Done()
//						err := f.v1UploadFile(ctx, o, buf, position, n, part, src.String(), "false")
//						if err != nil {
//							finished = true // 结束循环
//						}
//						<-ch
//					}(part, position, buf)
//				} else {
//					err = f.v1UploadFile(ctx, o, buf, position, n, part, src.String(), "true")
//					if err != nil {
//						finished = true // 结束循环
//					}
//				}
//			}
//
//			if f.opt.Version == "v2" {
//				if part != 0 {
//					ch <- struct{}{}
//					wg.Add(1)
//					go func(part, position int64, buf []byte) {
//						defer wg.Done()
//						err := f.v2RequestHttp(ctx, o, buf, "false", "", part, position, n)
//						if err != nil {
//							finished = true // 结束循环
//						}
//						<-ch
//					}(part, position, buf)
//				} else {
//					err = f.v2RequestHttp(ctx, o, buf, "true", "", part, position, n)
//					if err != nil {
//						finished = true // 结束循环
//					}
//				}
//			}
//		} else {
//			break
//		}
//		remaining -= n
//		position += n
//
//	}
//
//	wg.Wait()
//
//	return nil, hex.EncodeToString(hashmd5v2.Sum(nil))
//
//}

//func (f *Fs) fileMultipartUpload3(ctx context.Context, o *Object, in io.Reader, src fs.ObjectInfo, chunkSize fs.SizeSuffix) error {
//	size := src.Size()
//	remaining := size
//	position := int64(0) // 偏移量
//	part := int64(0)
//	var wg sync.WaitGroup
//	ch := make(chan struct{}, f.opt.NumberOfConcurrent) // 控制并发数量
//	//ch2 := make(chan string, 5)
//	defer close(ch)
//	count := 0
//
//	for {
//		if count != 0 {
//			break
//		}
//
//		if remaining > 0 {
//			n := int64(chunkSize)
//			if remaining < n {
//				n = remaining // 最后的块大小
//			}
//			// 将 in 分片
//			seg := readers.NewRepeatableReader(io.LimitReader(in, n))
//			MD5 := f.ReaderMd5(ctx, seg)
//			fmt.Println("第 %d 块， md5值 %s", part, MD5)
//			_, _ = seg.Seek(0, io.SeekStart)
//
//			if f.opt.Version == "v1" {
//				if part != 0 {
//					// 上传
//					wg.Add(1)
//					ch <- struct{}{}
//					go func(ctx context.Context, o *Object, seg io.ReadSeeker, position, n, part int64, name, reset string) {
//						defer wg.Done()
//						err := f.V1UploadFile(ctx, o, seg, position, n, part, name, reset)
//						if err != nil {
//							wg.Add(1)
//							go func(ctx context.Context, o *Object, seg io.ReadSeeker, position, n, part int64, name, reset string) {
//								defer wg.Done()
//								_, _ = seg.Seek(0, io.SeekStart)
//								err := f.V1UploadFile2(ctx, o, seg, position, n, part, name, reset)
//								if err != nil {
//									fs.Errorf(o, "文件需要重新上传。")
//									return
//								} else {
//									fs.Debugf(o, "第 %d 块 重传完成。", part)
//									<-ch
//								}
//							}(ctx, o, seg, position, n, part, src.String(), "false")
//
//						} else {
//							fs.Debugf(o, "第 %d 块 上传完成。", part)
//							<-ch
//						}
//
//					}(ctx, o, seg, position, n, part, src.String(), "false")
//				} else {
//					err := f.V1UploadFile(ctx, o, seg, position, n, part, src.String(), "true")
//					if err != nil {
//						return err
//					}
//					fs.Debugf(o, "第 %d 块 上传完成。", part)
//				}
//
//			}
//			if f.opt.Version == "v2" {
//				if part != 0 {
//					wg.Add(1) // 不能提前 因else中不属于go程 会站位置使程序无法退出
//					ch <- struct{}{}
//					go func(ctx context.Context, o *Object, seg io.ReadSeeker, reset, MD5 string, part, position, n int64, ch chan struct{}) {
//						defer wg.Done()
//						err := f.v2RequestHttp(ctx, o, seg, reset, MD5, part, position, n)
//						if err != nil {
//							wg.Add(1)
//							go func(ctx context.Context, o *Object, seg io.ReadSeeker, reset, MD5 string, part, position, n int64, ch chan struct{}) {
//								defer wg.Done()
//								_, _ = seg.Seek(0, io.SeekStart)
//								err := f.v2RequestHttp2(ctx, o, seg, reset, MD5, part, position, n)
//								if err != nil {
//									fs.Errorf(o, "文件需要重新上传。")
//									count = 1
//								} else {
//									fs.Debugf(o, "第 %d 块 重传完成。", part)
//									<-ch
//								}
//							}(ctx, o, seg, reset, MD5, part, position, n, ch)
//
//						} else {
//							fs.Debugf(o, "第 %d 块 上传完成。", part)
//							<-ch
//						}
//
//					}(ctx, o, seg, "false", MD5, part, position, n, ch)
//
//				} else {
//					err := f.v2RequestHttp(ctx, o, seg, "true", MD5, part, position, n)
//					if err != nil {
//						return err
//					}
//					fs.Debugf(o, "第 %d 块 上传完成。", part)
//				}
//
//			}
//			remaining -= n
//			position += n
//			part += 1
//		} else {
//			break
//		}
//
//	}
//
//	wg.Wait()
//	if count != 0 {
//		return errors.New("文件需要重新上传。")
//	} else {
//		return nil
//	}
//
//}
